package com.g.ocache.impl.index;

import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.commons.lang.builder.ToStringBuilder;
import org.apache.commons.lang.builder.ToStringStyle;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.g.ocache.Cache;
import com.g.ocache.event.CacheEvent;
import com.g.ocache.event.CacheEventPublisher;
import com.g.ocache.event.CacheEventType;
import com.g.ocache.index.Index;
import com.g.ocache.index.IndexBuilder;
import com.g.ocache.index.IndexLoader;
import com.g.ocache.index.Node;
import com.g.ocache.utils.CallableExecutor;


/**
 * 
 * @author zhangyf
 * @param <K>
 * @param <V>
 */
public abstract class AbstractIndex<K, V> implements Index<K, V> {
	//
	private static final Logger LOGGER = LoggerFactory.getLogger(AbstractIndex.class);
	
	//
	private final String name;
	private final AtomicBoolean verbose;
	private final AtomicBoolean started;
	private final ReentrantReadWriteLock lock;
	private final AtomicReference<Cache<K, V>> cache;
	private final CallableExecutor<Void> loaderExecutor;
	private final AtomicReference<IndexLoader<K, V>> loader;
	private final AtomicReference<IndexBuilder<K, V>> builder;
	
	//
	protected abstract void doStart() throws Exception;
	protected abstract void doStop() throws Exception;

	/**
	 * 
	 */
	public AbstractIndex(String name, IndexBuilder<K, V> builder, boolean fair) {
		this.name = name;
		this.verbose = new AtomicBoolean(true);
		this.started = new AtomicBoolean(false);
		this.lock = new ReentrantReadWriteLock(fair);
		this.cache = new AtomicReference<Cache<K, V>>();
		this.loader = new AtomicReference<IndexLoader<K, V>>();
		this.loaderExecutor = new CallableExecutor<Void>(false);
		this.builder = new AtomicReference<IndexBuilder<K, V>>(builder);
	}
	
	/**
	 * 
	 */
	public String toString() {
		return new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE)
		.append("name", name).toString();
	}
	
	/**
	 * 
	 */
	public final boolean isRunning() {
		return started.get();
	}
	
	public final void start() throws Exception {
		// Precondition checking
		if(!started.compareAndSet(false, true)) {
			throw new IllegalStateException("the cache has already been started");
		}
		
		//
		final IndexLoader<K, V> loader = getLoader();
		if(loader != null) {
			loader.start();
		}
		
		//
		doStart();
	}
	
	public final void stop() throws Exception {
		// Precondition checking
		if(!started.compareAndSet(true, false)) {
			return;
		}
		
		//
		doStop();
		
		//
		final IndexLoader<K, V> loader = getLoader();
		if(loader != null) {
			loader.stop();
		}
	}
	
	/**
	 * 
	 */
	public final String getName() {
		return name;
	}
	
	public final boolean isFair() {
		return lock.isFair();
	}
	
	public final boolean isVerbose() {
		return verbose.get();
	}
	
	public final void setVerbose(boolean verbose) {
		this.verbose.set(verbose);
	}
	
	public final Cache<K, V> getCache() {
		return cache.get();
	}

	public final void setCache(Cache<K, V> cache) {
		this.cache.set(cache);
	}
	
	public final IndexLoader<K, V> getLoader() {
		return loader.get();
	}
	
	public final void setLoader(IndexLoader<K, V> loader) {
		this.loader.set(loader);
	}
	
	public final IndexBuilder<K, V> getBuilder() {
		return builder.get();
	}
	
	public void lock(boolean exclusive) {
		final Lock l = exclusive ? this.lock.writeLock() : this.lock.readLock();
		l.lock();
	}
	
	public void unlock(boolean exclusive) {
		final Lock l = exclusive ? this.lock.writeLock() : this.lock.readLock();
		l.unlock();
	}
	
	/**
	 * 
	 */
	protected void load(Set<Node<K, V>> nodes) {
		//
		if(nodes == null || nodes.size() == 0) {
			return;
		}
		final IndexLoader<K, V> loader = getLoader();
		if(loader == null) {
			return;
		}
		
		//
		for(Node<K, V> node : nodes) {
			//
			if(node.getId() == null) { // root
				LOGGER.warn("assertion failed, should not reach here, index: {}", getName());
				continue;
			}
			if(node.isInitialized()) {
				LOGGER.warn("node has already been initialized, index: {}, path: ", getName(), node.getIdPath());
				continue;
			}
			
			//
			StringBuilder id = new StringBuilder();
			for(Iterator<Object> iter = node.getIdPath().iterator(); iter.hasNext(); ) {
				id.append(System.identityHashCode(iter.next()));
				if(iter.hasNext()) {
					id.append("-");
				}
			}
			
			//
			try {
				loaderExecutor.execute(id.toString(), new LoadingCallable(node, loader));
			} catch(Exception e) {
				// If loader threw an exception, that exception can be found in "indexedLoad" event
				if(isVerbose()) {
					LOGGER.error("lazy load failed, index: " + getName() + ", path: " + node.getIdPath(), e);
				}
			}
		}
	}
	
	protected class LoadingCallable implements Callable<Void> {
		//
		private final Node<K, V> node;
		private final IndexLoader<K, V> loader;
		
		/**
		 * 
		 */
		public LoadingCallable(Node<K, V> node, IndexLoader<K, V> loader) {
			this.node = node;
			this.loader = loader;
		}

		/**
		 * 
		 */
		public Void call() throws Exception {
			Exception exception = null;
			try {
				//
				if(isVerbose() && LOGGER.isInfoEnabled()) {
					LOGGER.info("lazy load, index: {}, path: {}", getName(), node.getIdPath());
				}
				
				//
				Map<K, V> values = this.loader.load(node.getIdPath());
				
				//
				if(isVerbose() && LOGGER.isInfoEnabled()) {
					LOGGER.info("lazy load finished, index: {}, count: {}", getName(), (values == null ? -1 : values.size()));
				}
				
				//
				getCache().putAll(values);
				
				//
				node.setInitialized(true);
				return null;
			} catch(Exception e) {
				exception = e;
				throw e;
			} finally {
				// Publish "indexedLoad" event
				final CacheEvent<K, V> event = new CacheEvent<K, V>(CacheEventType.indexedLoad);
				event.setException(exception);
				event.setDescription(node.getIdPath().toString());
				CacheEventPublisher<K, V> publisher = getCache().getPublisher();
				if(publisher != null) {
					publisher.publish(event);
				}
			}
		}
	}
}
